package net.bwie.realtime.jtp.utils;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class jianbiao {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建创表环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql(
                "CREATE TABLE douyin_live_business (\n" +
                "  `orderId` STRING,\n" +
                "  `eventTime` STRING,\n" +
                "  `roomId` STRING,\n" +
                "  `userId` STRING,\n" +
                "  `productSku` STRING,\n" +
                "  `orderAmount` DOUBLE,\n" +
                "  `payAmount` STRING,\n" +
                "  `payStatus` STRING,\n" +
                "  `anchorId` STRING,\n" +
                "  `roomType` STRING,\n" +
                "  `startTime` STRING,\n" +
                "  `endTime` STRING,\n" +
                "  `interactId` STRING,\n" +
                "  `interactEventType` STRING,\n" +
                "  `interactType` STRING,\n" +
                "  `content` STRING,\n" +
                "  `visitId` STRING,\n" +
                "  `enterTime` STRING,\n" +
                "  `leaveTime` STRING,\n" +
                "  `source` STRING,\n" +
                "  `duration` INT,\n" +
                "  `productName` STRING,\n" +
                "  `price` DOUBLE,\n" +
                "  `category` STRING,\n" +
                "  `status` STRING \n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'douyin_live_business',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");
        tEnv.executeSql("select * from douyin_live_business limit 10").print();

        Table deal = getDeal(tEnv);

        //指标
        getNum(tEnv, deal);

        //写入Doris
        saveDoris(tEnv, deal);




//        // 启动
//        env.execute("jianbiao");
    }

    /**
     *
     */
    private static void saveDoris(StreamTableEnvironment tEnv, Table deal) {

        tEnv.executeSql(
                "INSERT INTO 表明 \n"
                + "SELECT * FROM deal"
        );


    }

    /**
     *
     */
    private static void getNum(StreamTableEnvironment tEnv, Table deal) {
        //建表
        tEnv.createTemporaryView("deal", deal);



    }

    /**
     *
     * @param tEnv
     */
    private static Table getDeal(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery(
                "select * from douyin_live_business limit 10"
        );

        return table;

    }


}
